package com.desheng.bigdata.flink.window

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * 增量窗口函数操作
  *     每当窗口中新加入一条记录，就会参与聚合计算操作
  *     常见的就是reduce、aggregate操作
  */
object _03IncrementWindowOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment

        val lines = env.socketTextStream("bigdata01", 9999)

        lines.flatMap(_.split("\\s+"))
            .map((_, 1))
            .keyBy(0)
            .timeWindow(Time.seconds(4))
            .reduce(new ReduceFunction[(String, Int)](){
                override def reduce(v1: (String, Int), v2: (String, Int)): (String, Int) = {
                    (v1._1, v1._2 + v2._2)
                }
            })
            .print()


        env.execute()
    }
}
